package pl.com.qapps.datastore.berkeley;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

import pl.com.qapps.datastore.base.FutureHelper;
import pl.com.qapps.datastore.base.TransactionStack;
import pl.com.qapps.datastore.base.FutureHelper.CumulativeAggregateFuture;
import pl.com.qapps.datastore.berkeley.impl.TransactionImpl;


import com.google.appengine.api.datastore.AccessibilityHelper;
import com.google.appengine.api.datastore.AsyncDatastoreService;
import com.google.appengine.api.datastore.DatastoreAttributes;
import com.google.appengine.api.datastore.DatastoreServiceConfig;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Index;
import com.google.appengine.api.datastore.Index.IndexState;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyRange;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.api.datastore.TransactionOptions;
import com.google.appengine.api.utils.FutureWrapper;
import com.google.appengine.repackaged.com.google.common.base.Pair;
import com.google.appengine.repackaged.com.google.common.collect.AbstractIterator;
import com.google.apphosting.api.DatastorePb.GetRequest;

/**
 * 
 * @author Jakub Dykowski
 *
 */
public final class AsyncBerkeleyDatastoreService extends
		BaseBerkeleyDatastoreService implements AsyncDatastoreService {

	AsyncBerkeleyDatastoreService(DatastoreServiceConfig cfg, FileStore store,
			TransactionStack stack, ExecutorService dispatcher) {
		super(cfg, store, stack, dispatcher);
	}

	/***
	 * An aggregate future that uses an iterator to match results to requested
	 * elements.
	 * 
	 * @param <K>
	 *            response type
	 * @param <I>
	 *            type being iterated over
	 * @param <V>
	 *            result type
	 */
	private abstract static class IteratingAggregateFuture<K, I, V> extends
			CumulativeAggregateFuture<K, Pair<Iterator<I>, V>, V> {
		public IteratingAggregateFuture(Iterable<Future<K>> futures) {
			super(futures);
		}

		protected abstract V aggregate(K intermediateResult,
				Iterator<I> iterator, V result);

		protected abstract Iterator<I> initIterator();

		protected abstract V initResult();

		@Override
		protected final Pair<Iterator<I>, V> aggregate(K intermediateResult,
				Pair<Iterator<I>, V> result) {
			return Pair.of(result.first,
					aggregate(intermediateResult, result.first, result.second));
		}

		@Override
		protected V finalizeResult(Pair<Iterator<I>, V> result) {
			return result.second;
		}

		@Override
		protected final Pair<Iterator<I>, V> initIntermediateResult() {
			return Pair.of(initIterator(), initResult());
		}
	}

	/**
	 * Models an item and its associated index in some ordered collection.
	 * 
	 * @param <T>
	 *            The type of the item.
	 */
	static class IndexedItem<T> implements Comparable<IndexedItem<T>> {
		final T item;
		final int index;

		IndexedItem(T item, int index) {
			this.item = item;
			this.index = index;
		}

		@Override
		public int compareTo(IndexedItem other) {
			return Integer.valueOf(index).compareTo(other.index);
		}
	}

	/**
	 * {@link Iterable implementation that converts a Iterable<IndexedItem<T>>
	 * to a Iterable<T>.
	 */
	private static class UnwrappingIterable<T> implements Iterable<T> {
		private final Iterable<IndexedItem<T>> innerIterable;

		private UnwrappingIterable(Iterable<IndexedItem<T>> innerIterable) {
			this.innerIterable = innerIterable;
		}

		@Override
		public Iterator<T> iterator() {
			return new AbstractIterator<T>() {
				Iterator<IndexedItem<T>> inner = innerIterable.iterator();

				@Override
				protected T computeNext() {
					if (inner.hasNext()) {
						return inner.next().item;
					}
					endOfData();
					return null;
				}
			};
		}
	}

	/**
	 * {@link CumulativeAggregateFuture} implementation that operates on Futures
	 * that return List<IndexedItem<Key>>. The final result produced by this
	 * Future returns the Keys sorted by their indexes.
	 */
	private static class SortingAggregateFuture
			extends
			CumulativeAggregateFuture<List<IndexedItem<Key>>, List<IndexedItem<Key>>, List<Key>> {
		private SortingAggregateFuture(
				Iterable<Future<List<IndexedItem<Key>>>> futures) {
			super(futures);
		}

		@Override
		protected List<IndexedItem<Key>> initIntermediateResult() {
			return new ArrayList<IndexedItem<Key>>();
		}

		@Override
		protected List<IndexedItem<Key>> aggregate(
				List<IndexedItem<Key>> intermediateResult,
				List<IndexedItem<Key>> result) {
			intermediateResult.addAll(result);
			return intermediateResult;
		}

		@Override
		protected List<Key> finalizeResult(
				List<IndexedItem<Key>> unorderedResult) {
			Collections.sort(unorderedResult);
			List<Key> orderedResult = new ArrayList<Key>(unorderedResult.size());
			for (IndexedItem<Key> key : unorderedResult) {
				orderedResult.add(key.item);
			}
			return orderedResult;
		}
	}

	@Override
	public Future<KeyRange> allocateIds(String kind, long num) {
		return allocateIds(null, kind, num);
	}

	@Override
	public Future<KeyRange> allocateIds(final Key parent, final String kind,
			final long num) {
		// System.out.println("allocating id: " + parent + " " + kind + " " +
		// num);
		if (parent != null && !parent.isComplete())
			throw new IllegalArgumentException("parent cannot be incomplete");
		if (kind == null)
			throw new IllegalArgumentException("kind cannot be null");
		if (num < 1 || num > Long.MAX_VALUE - 1)
			throw new IllegalArgumentException("num must be between 0 and "
					+ (Long.MAX_VALUE - 1));

		AlocateIdsRequest req = new AlocateIdsRequest();
		req.setAmount(num);

		return new FutureWrapper<Long, KeyRange>(runInTransaction(req)) {

			@Override
			protected Throwable convertException(Throwable cause) {
				return cause;
			}

			@Override
			protected KeyRange wrap(Long lastId) throws Exception {
				return new KeyRange(parent, kind, lastId.longValue() - num + 1,
						lastId);
			}

		};
	}

	@Override
	public Future<Transaction> beginTransaction() {
		return beginTransaction(null);
	}

	@Override
	public Future<Transaction> beginTransaction(final TransactionOptions options) {
		return async(new Callable<Transaction>() {

			@Override
			public Transaction call() throws Exception {
				return beginTransactionInternal(options != null ? options
						: TransactionOptions.Builder.withDefaults());
			}

		});
	}

	@Override
	public Future<Void> delete(Key... keys) {
		return delete(null, Arrays.asList(keys));
	}

	@Override
	public Future<Void> delete(Iterable<Key> keys) {
		return delete(null, keys);
	}

	@Override
	public Future<Void> delete(Transaction txn, Key... keys) {
		return delete(txn, Arrays.asList(keys));
	}

	@Override
	public Future<Void> delete(Transaction txn, Iterable<Key> keys) {
		verifyKeys(keys);

		DeleteRequest req = new DeleteRequest();
		req.setTxn(txn);
		req.setKeys(keys);

		return runInTransaction(req);
	}

	@Override
	public Future<Entity> get(Key key) {
		return get(null, key);
	}

	@Override
	public Future<Map<Key, Entity>> get(Iterable<Key> keys) {
		return get(null, keys);
	}

	@Override
	public Future<Entity> get(Transaction txn, final Key key) {
		return new FutureWrapper<Map<Key, Entity>, Entity>(get(txn,
				Arrays.asList(key))) {

			@Override
			protected Throwable convertException(Throwable cause) {
				return cause;
			}

			@Override
			protected Entity wrap(Map<Key, Entity> map) throws Exception {
				return map.get(key);
			}
		};
	}

	@Override
	public Future<Map<Key, Entity>> get(final Transaction txn,
			final Iterable<Key> keys) {
		verifyKeys(keys);
		verifyTxn(txn);
		TransactionImpl.ensureTxnActive(txn);

		GetRequest req = new GetRequest();
		for (Key key : keys) {
			req.addKey(AccessibilityHelper.convertToPb(key));
		}

		return async(new Callable<Map<Key, Entity>>() {

			@Override
			public Map<Key, Entity> call() throws Exception {
				Iterable<Entity> entities = store.get(keys,
						(TransactionImpl) txn);
				Map<Key, Entity> map = new HashMap<>();
				for (Entity entity : entities) {
					map.put(entity.getKey(), entity);
				}
				return map;
			}

		});

		// GetRequestDep req = new GetReques();
		// req.setKeys(keys);
		// req.setTxn(txn);
		//
		// return runInTransaction(req);
	}

	@Override
	public Future<DatastoreAttributes> getDatastoreAttributes() {
		throw new UnsupportedOperationException("not supported yet");
	}

	@Override
	public Future<Map<Index, IndexState>> getIndexes() {
		throw new UnsupportedOperationException("not supported yet or ever");
	}

	@Override
	public Future<Key> put(Entity entity) {
		return put(null, entity);
	}

	@Override
	public Future<List<Key>> put(Iterable<Entity> entities) {
		return put(null, entities);
	}

	@Override
	public Future<Key> put(Transaction txn, Entity entity) {
		return new FutureWrapper<List<Key>, Key>(
				put(txn, Arrays.asList(entity))) {

			@Override
			protected Throwable convertException(Throwable cause) {
				return cause;
			}

			@Override
			protected Key wrap(List<Key> keys) throws Exception {
				return keys.get(0);
			}

		};
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.google.appengine.api.datastore.AsyncDatastoreService#put(com.google
	 * .appengine.api.datastore.Transaction, java.lang.Iterable)
	 */
	@Override
	public Future<List<Key>> put(Transaction txn, Iterable<Entity> entities) {

		verifyTxn(txn);
		TransactionImpl.ensureTxnActive(txn);

		for (Entity entity : entities) {
			if (!entity.getKey().isComplete()) {
				entities = generateIds(entities);
				break;
			}
		}

		PutRequestDep req = new PutRequestDep();
		req.setTxn(txn);
		req.setEntities(entities);

		return runInTransaction(req);
	}

	private List<Entity> generateIds(Iterable<Entity> entities) {
		List<Entity> result = new LinkedList<>();
		for (Entity entity : entities) {
			if (!entity.getKey().isComplete()) {
				Key generated = FutureHelper.quietGet(
						allocateIds(entity.getParent(), entity.getKind(), 1L))
						.getStart();
				Entity newEntity = new Entity(generated);
				newEntity.setPropertiesFrom(entity);
				result.add(newEntity);
			} else
				result.add(entity);
		}
		return result;
		// Map<Pair<Key, String>, List<Entity>> toGenerate = new HashMap<>();
		// List<Entity> result = new LinkedList<>();
		// for (Entity entity : entities) {
		// if (!entity.getKey().isComplete()) {
		// Pair<Key, String> pair = Pair.of(entity.getParent(),
		// entity.getKind());
		// List<Entity> common = toGenerate.get(pair);
		// if (common == null) {
		// common = new LinkedList<>();
		// toGenerate.put(pair, common);
		// }
		// common.add(entity);
		// } else result.add(entity);
		// }
		// for(Entry<Pair<Key, String>, List<Entity>> entry :
		// toGenerate.entrySet()) {
		// for(Key key : allocateIds(entry.getKey().first,
		// entry.getKey().second,entry.getValue().size())) {
		//
		// }
		// }
	}

	private <T> Future<T> async(Callable<T> task) {
		return dispatcher.submit(task);
	}

	private <T> Future<T> asyncCall(Callable<T> task) {
		return new FutureWrapper<T, T>(async(task)) {

			@Override
			protected Throwable convertException(Throwable cause) {
				System.out.println("Wrapping exception from ascyn call: "
						+ cause);
				return cause;
			}

			@Override
			protected T wrap(T value) throws Exception {
				return value;
			}

		};
	}

	private <T> Future<T> runInTransaction(final BaseRequest<T> req) {
		req.init(store);
		return asyncCall(new TxnRunner<T>(req.getTxn()) {

			@Override
			T runInternal(com.sleepycat.je.Transaction txn) {
				return req.run(store, txn);
			}

		});

	}

	private static void verifyKeys(Iterable<Key> keys) {
		for (Key key : keys) {
			if (key == null)
				throw new IllegalArgumentException("one of given keys is null");
			if (!key.isComplete()) {
				throw new IllegalArgumentException("key '" + key
						+ "' is incomplete");
			}
		}
	}
}
